package com.donglink.test.kafkaandstorm;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import redis.clients.jedis.Jedis;

import java.util.HashMap;
import java.util.Map;

public class CountBolt extends BaseRichBolt {

    OutputCollector outputCollector ;
    TopologyContext topologyContext;
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

        this.outputCollector=outputCollector;
        this.topologyContext=topologyContext;
    }

    @Override
    public void execute(Tuple tuple) {

        String word=tuple.getStringByField("word");
        int count=tuple.getIntegerByField("count");

        mapToRedis(word,count);


    }

    private static void mapToRedis(String word,int count)
    {
        Jedis jedis=RedisPool.getJedis();

        if(jedis.exists(word))
        {
            int tmp=Integer.parseInt(jedis.get(word));
            jedis.set(word,String.valueOf(tmp+count));
            System.out.println("www");
        }
        else
        {
            jedis.append(word,String.valueOf(count));
            System.out.println("qqq");
        }
        jedis.close();
    }


    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
